package cn.azzhu.day07

import cn.azzhu.day02.SensorSource
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api._
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.table.api.bridge.scala.{StreamTableEnvironment, tableConversions}
import org.apache.flink.types.Row
/**
 * Flink-table-SQL：时间特性
 * @author azzhu
 * @create 2020-09-23 22:07:12
 */
object TableProTime {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val settings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    val tEnv = StreamTableEnvironment.create(env, settings)

    val stream = env.addSource(new SensorSource)

    //pt.proctime指定了处理时间是pt字段，必须放在最后
   val table =  tEnv.fromDataStream(stream,'id,'timestamp as 'ts,'temperature as 'temp,'pt.proctime)

    //table API窗口操作
    table
      //开窗口，并命名为'w
      .window(Tumble over 10.second() on 'pt as 'w)
      // .keyBy(_.id).timeWindow(Time.seconds(10))
      .groupBy('id,'w)
      //.porcess
      .select('id,'id.count())
      .toRetractStream[Row]
      .print()

    env.execute("SQLProTime")
  }
}
